#include <stdio.h>
#include <stdlib.h>
#include <semaphore.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <unistd.h>
#include <string.h>
#include <errno.h>
#include <sys/mman.h>
#include <sys/time.h>
#include <pthread.h>
#include <string.h>

#define MAX_QUEUE (20)
#define INIT_FLAG (0xA5A5A5A5)
// 队列项头部固定字节存储实际大小
#define ITEM_HDR (2)
#define MIN(a, b) ((a) < (b) ? (a) : (b))

typedef struct
{
    char name[64];
    // 队列初始化标志
    unsigned int init_flag;
    // 当前队列可读元素序号
    unsigned int pos_r;
    // 当前队列可写元素序号
    unsigned int pos_w;
    // 队列中元素最大存储大小
    unsigned int size;
    // 队列中元素个数
    unsigned int num;
    // 资源可读信号量，用于同步
    sem_t sem_can_read;
    // 资源可写信号量，用于同步
    sem_t sem_can_write;
    // 共享内存
    unsigned char *data;
} queue_t;

static queue_t g_queues[MAX_QUEUE];
static pthread_mutex_t g_lock = PTHREAD_MUTEX_INITIALIZER;

static int delayms(int ms)
{
    int ret;
    struct timeval timeout;
    timeout.tv_sec = ms / 1000;
    timeout.tv_usec = (ms % 1000) * 1000;
    do
    {
        ret = select(0, NULL, NULL, NULL, &timeout);
        if (ret < 0 && errno == EINTR)
        {
            continue;
        }
        else if (ret < 0)
        {
            return -1;
        }
    } while (ret != 0);
    return 0;
}

// 按名字查找空闲的队列项，一般是发送者使用
static int thread_queue_find_free(char *name)
{
    int i;
    for (i = 0; i < MAX_QUEUE; i++)
    {
        if (strcmp(g_queues[i].name, name) == 0)
        {
            return -1;
        }
    }

    for (i = 0; i < MAX_QUEUE; i++)
    {
        if (g_queues[i].init_flag != INIT_FLAG)
        {
            return i;
        }
    }
    return -2;
}

// 按名字查找可用的队列项，一般都接收者使用
static int thread_queue_find_inited(char *name)
{
    int i;
    for (i = 0; i < MAX_QUEUE; i++)
    {
        if (g_queues[i].init_flag == INIT_FLAG &&
            strcmp(g_queues[i].name, name) == 0)
        {
            return i;
        }
    }

    return -1;
}

// 按名字创建消息队列，一般是发送者使用
int thread_queue_create(char *name, int size, int num)
{
    int total_size = 0;
    int queue_id, ret = 0;
    queue_t *queue;

    pthread_mutex_lock(&g_lock);
    queue_id = thread_queue_find_free(name);
    if (queue_id < 0)
    {
        ret = -1;
        goto failed;
    }

    queue = &g_queues[queue_id];
    sem_init(&queue->sem_can_read, 0, 0);
    sem_init(&queue->sem_can_write, 0, num);

    total_size = (size + ITEM_HDR) * num;
    queue->data = malloc(total_size);
    if (queue->data == NULL)
    {
        ret = -2;
        goto failed;
    }
    strcpy(queue->name, name);
    queue->pos_r = 0;
    queue->pos_w = 0;
    queue->init_flag = INIT_FLAG;
    queue->size = size;
    queue->num = num;
    ret = queue_id;
failed:
    if (queue->data != NULL)
    {
        free(queue->data);
    }
    pthread_mutex_unlock(&g_lock);
    return ret;
}

// 按名字打开消息队列，一般是接收者使用
int thread_queue_open(char *name, int timeout)
{
    int queue_id;
    int elapsed_ms = 0;
    queue_t *queue;

    while (elapsed_ms < timeout)
    {
        queue_id = thread_queue_find_inited(name);
        if (queue_id >= 0)
        {
            return queue_id;
        }
        delayms(20);
        elapsed_ms += 20;
    }
    return -1;
}

// 发送消息，一般是发送者使用
int thread_queue_send(int queue_id, char *buffer, int size)
{
    char *item;
    queue_t *queue = &g_queues[queue_id];
    // 可写资源-1
    sem_wait(&queue->sem_can_write);

    item = queue->data + queue->pos_w * (queue->size + ITEM_HDR);
    size = MIN(size, queue->size);
    item[0] = (size >> 8) & 0xff;
    item[1] = size & 0xff;
    memcpy(item + ITEM_HDR, buffer, size);
    queue->pos_w = (queue->pos_w + 1) % queue->num;

    // 可读资源+1
    sem_post(&queue->sem_can_read);
    return size;
}

// 接收消息，一般是接收者使用
int thread_queue_recv(int queue_id, char *buffer, int size)
{
    char *item;
    int real_size = 0;
    queue_t *queue = &g_queues[queue_id];
    // 可读资源-1
    sem_wait(&queue->sem_can_read);

    item = queue->data + queue->pos_r * (queue->size + ITEM_HDR);
    real_size = (item[0] << 8) + item[1];
    size = MIN(size, real_size);
    memcpy(buffer, item + ITEM_HDR, size);
    queue->pos_r = (queue->pos_r + 1) % queue->num;

    // 可写资源+1
    sem_post(&queue->sem_can_write);

    return size;
}

// 关闭队列
void thread_queue_close(int queue_id)
{
    queue_t *queue = &g_queues[queue_id];
    pthread_mutex_lock(&g_lock);

    if (queue->init_flag == INIT_FLAG)
    {
        sem_close(&queue->sem_can_read);
        sem_close(&queue->sem_can_write);
        if (queue->data)
        {
            free(queue->data);
        }
        memset(queue, 0x00, sizeof(queue_t));
    }

    pthread_mutex_unlock(&g_lock);
    return;
}